package net.xuele.learn.rsocket;

import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import net.xuele.learn.rsocket.support.DataPublisher;
import net.xuele.learn.rsocket.support.GameController;
import org.apache.http.client.utils.DateUtils;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Date;
import java.util.concurrent.CountDownLatch;

/**
 * @author: patrick-Hikari
 * @description:
 * @date: 2021/8/18 5:41 下午
 */
public class Server {
    private final Disposable server;

    private final DataPublisher dataPublisher = new DataPublisher();

    private final GameController gameController;

    public Server() {
        this.server = RSocketFactory.receive()
                .acceptor((setupPayload,reactiveSocket)-> Mono.just(new RSocketImpl()))
                .transport(TcpServerTransport.create("localhost",8070))
                .start()
                .subscribe();

        this.gameController = new GameController("server player");
    }


    public void dispose() {
        dataPublisher.complete();
        this.server.dispose();
    }

    /**
     * 处理客户端请求
     */
    private class RSocketImpl extends AbstractRSocket {
        /**
         * 请求响应模型
         *
         * 对应的客户端是ReqResClient，发送的消息包含在了payload中
         * @param payload 包含消息体和元数据，被各种交互模型所使用。其内容是二进制的，但有一些方法支持字符串形式的内容
         * @return Mono实现了org.reactivestreams.Publisher 接口，代表0到1个元素的发布者
         */
        @Override
        public Mono<Payload> requestResponse(Payload payload) {
            try {
                String dataUtf8 = payload.getDataUtf8();
                System.out.println("-----"+dataUtf8);
                return Mono.just(DefaultPayload.create("i received "+ dataUtf8));
            } catch (Exception e) {
                return Mono.error(e);
            }
        }

        /**
         * 在当前的场景下，服务端收到消息之后就将其发布出去，而服务端的另一个线程则将这个数据当作来源，返回给对应的客户端
         * @param payload
         * @return
         */
        @Override
        public Mono<Void> fireAndForget(Payload payload) {
            try {
                // 需要发布消息
                dataPublisher.publish(payload);
                return Mono.empty();
            } catch (Exception e) {
                return Mono.error(e);
            }
        }

        /**
         * 请求流，对数据流（通常是无限的）进行操作
         * 请求方将单个帧发送到响应方，并获取数据流。这种交互方式使服务能够从"拉数据"切换为"推数据"策略；无需向响应着发送定期请求，
         * 请求方可以订阅流并对收到的数据做出反应（数据可用时，将自动到达）
         *
         * 一个典型的场景是：请求方是云服务中的微服务，而响应方是温度传感器，它会将温度数据发送给微服务。
         *
         * 这个过程中请求方请求一次之后，就会源源不断的收到响应数据
         *
         *
         * server端处理请求的方法是io.rsocket.RSocketServer#handleFrame
         * @param payload
         * @return Flux实现了org.reactivestreams.Publisher接口，代表0到N个元素的发布者
         */
        @Override
        public Flux<Payload> requestStream(Payload payload) {
            String data = payload.getDataUtf8();
            // 收到了来自客户端的请求之后，将会源源不断的向客户端推送数据
            if ("data".equals(data)) {
                System.out.println("getting msg from me..." + Thread.currentThread().getName());
                // 数据来自于dataPublisher发布者
//                Flux.just(DefaultPayload.create("test"));
                return Flux.from(dataPublisher);
            }
            return Flux.error(new IllegalStateException(data));
        }

        /**
         * 入参出参都是publisher--payload stream
         *
         * RSocket 可以使用单个物理连接将数据从请求方传输到响应方，反之亦然。当请求方更新订阅时（如，更改订阅规则），
         * 这种交互方式可能很有用。如果没有双向通道，客户端将不得不取消流并使用新参数重新请求它
         * @param payloads
         * @return
         */
        @Override
        public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
            Flux.from(payloads)
                    .subscribe(gameController::processPayload);
            // 这里调用的subscribe方法是将
            return Flux.from(gameController);
        }
    }
}
